package com.jiang.admin.core.thread;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.web.client.RestTemplate;

import com.jiang.admin.core.config.JobAdminConfig;
import com.jiang.admin.core.model.TriggerParam;
import com.jiang.admin.entity.JobLog;
import com.jiang.common.ResultTO;
import com.jiang.common.StatusEnum;

/**
 * 任务处理线程
 * @author jiangyougan 2018年12月2日 下午6:28:21 
 *
 */
public class JobThread extends Thread {
	private static Logger logger = LoggerFactory.getLogger(JobThread.class);

	private int jobId;
	private LinkedBlockingQueue<TriggerParam> triggerQueue;
	private Set<Integer> triggerLogIdSet;

	//volatile 共享变量多线程可见
	private volatile boolean toStop = false;
	private int idleTimes = 0;

	public JobThread(int jobId) {
		this.jobId = jobId;
		this.triggerQueue = new LinkedBlockingQueue<TriggerParam>();
		this.triggerLogIdSet = Collections.synchronizedSet(new HashSet<Integer>());
	}

	/**
	 * @param triggerParam
	 * @return
	 */
	public ResultTO<String> pushTriggerQueue(TriggerParam triggerParam) {
		// 去重
		if (triggerLogIdSet.contains(triggerParam.getLogId())) {
			logger.info(">>>>>>>>>>> JobThread triggerLogIdSet repeate trigger job, logId:{}", triggerParam.getLogId());
			return ResultTO.fail(StatusEnum.BUSINESS, "repeate trigger job, logId:" + triggerParam.getLogId());
		}
		triggerLogIdSet.add(triggerParam.getLogId());
		triggerQueue.add(triggerParam);
		return ResultTO.success("");
	}
	
	public void toStop() {
		/**
		 * Thread.interrupt只支持终止线程的阻塞状态(wait、join、sleep)，
		 * 在阻塞出抛出InterruptedException异常,但是并不会终止运行的线程本身；
		 * 所以需要注意，此处彻底销毁本线程
		 */
		this.toStop = true;
	}

	@SuppressWarnings("unchecked")
	@Override
	public void run() {
		// 执行任务线程
		while (!toStop) {
			idleTimes++;
			
			TriggerParam triggerParam = null;
			ResultTO<String> executeResult = null;
			try {
				// 从队列中取数据
				triggerParam = triggerQueue.poll(3L, TimeUnit.SECONDS);
				if (triggerParam != null) {
					triggerLogIdSet.remove(triggerParam.getLogId());
					//获取远程调用客户端
					RestTemplate restTemplate = JobAdminConfig.getAdminConfig().getRestTemplate();
					//格式化远程调用地址和参数
					String url = String.format("http://%s/%s?%s", triggerParam.getServiceId(), triggerParam.getPath(), triggerParam.getExecutorParams());
					logger.info(">>>>>>>>>>> JobThread url:"+ url + ", logId: " + triggerParam.getLogId());
					long start = System.currentTimeMillis();
					//发起请求
					executeResult = restTemplate.getForObject(url, ResultTO.class);
					long end = System.currentTimeMillis();
					logger.info(">>>>>>>>>>> JobThread executeResult:"+ executeResult.toString()+ ", logId: " + triggerParam.getLogId() +", time:" +(end - start));
				} else {
					//循环10次，如果一直没有任务，则关闭线程
					if (idleTimes > 10) {
						JobThreadManager.removeJobThread(jobId);
					}				
				}
			} catch (Throwable e) {
				if (toStop) {
					logger.info(">>>>>>>>>>> JobThread stop: toStop is true");
				}
				StringWriter stringWriter = new StringWriter();
				e.printStackTrace(new PrintWriter(stringWriter));
				String errorMsg = stringWriter.toString();
				executeResult = ResultTO.fail(StatusEnum.BUSINESS, errorMsg);
				logger.error(">>>>>>>>>>> JobThread Exception:" + errorMsg + " job execute end(error), logId: " + triggerParam.getLogId());
			} finally {
				if (triggerParam != null) {
					JobLog jobLog = new JobLog();
					jobLog.setId(triggerParam.getLogId());
					jobLog.setHandleTime(new Date());
					if (!toStop) {
                        // 正常结束任务
						jobLog.setHandleCode(Integer.valueOf(executeResult.getCode()));
						jobLog.setHandleMsg(executeResult.getMsg());
                    } else {
                        // 任务线程中断
                    	ResultTO<String> stopResult = ResultTO.fail(StatusEnum.BUSINESS,  "job stoped");
                    	jobLog.setHandleCode(Integer.valueOf(stopResult.getCode()));
						jobLog.setHandleMsg(stopResult.getMsg());
                    }
					JobAdminConfig.getAdminConfig().getJobLogDao().updateHandleInfo(jobLog);
				}
			}
		}

		// 当线程结束且队列中任然有未执行任务
		while(triggerQueue !=null && triggerQueue.size()>0){
			TriggerParam triggerParam = triggerQueue.poll();
			if (triggerParam!=null) {
				logger.error(">>>>>>>>>>> JobThread job not executed, logId:" + triggerParam.getLogId());
				ResultTO<String> stopResult = ResultTO.fail(StatusEnum.BUSINESS,  "job not executed in the job queue, logId:" + triggerParam.getLogId());
				JobLog jobLog = new JobLog();
				jobLog.setId(triggerParam.getLogId());
				jobLog.setHandleTime(new Date());
				jobLog.setHandleCode(Integer.valueOf(stopResult.getCode()));
				jobLog.setHandleMsg(stopResult.getMsg());
				JobAdminConfig.getAdminConfig().getJobLogDao().updateHandleInfo(jobLog);
			}
		}

		logger.info(">>>>>>>>>>> JobThread stoped threadName:" + Thread.currentThread().getName());
	}
}
